package com.whz.disruptor.handler;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;
import lombok.extern.slf4j.Slf4j;
import com.whz.disruptor.event.MessageEvent;

import java.util.function.Consumer;

/**
 * disruptor 消费者实现
 *
 * @author whz
 */
@Slf4j
public class ConsumerHandler implements EventHandler<MessageEvent>, WorkHandler<MessageEvent> {

    public static final ConsumerHandler DEFAULT_CONSUMER_HANDLER = new ConsumerHandler(msg -> log.info("consume:{}", msg));

    private final Consumer<Object> consumer;

    public ConsumerHandler(Consumer<Object> consumer) {
        this.consumer = consumer;
    }

    @Override
    public void onEvent(MessageEvent messageEvent, long l, boolean b) {
        log.debug("EventHandler -> consume message event:{}", messageEvent);
        this.consumer.accept(messageEvent.getMessage());
    }

    @Override
    public void onEvent(MessageEvent tMessageEvent) {
        log.debug("WorkHandler -> consume message event:{}", tMessageEvent);
        this.consumer.accept(tMessageEvent.getMessage());
    }
}